package com.huitone.smspfm.interfacz.mo.report;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.huitone.smspfm.interfacz.logger.ExtLogger;
import com.huitone.smspfm.interfacz.mo.SvrNodeClientsManager;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;

import com.bin.common.utils.DateTimeUtils;
import com.bin.common.utils.StringUtils;
import com.huitone.smspfm.interfacz.core.AppStaticConfigs;
import com.huitone.smspfm.interfacz.dao.entity.ErrorMng;
import com.huitone.smspfm.interfacz.dao.entity.SvrNodeDef;
import com.huitone.smspfm.interfacz.model.NotifyHash;
import com.huitone.smspfm.interfacz.model.SmsReportForResend;
import com.huitone.smspfm.redis.model.AppCanstant;
import com.huitone.smspfm.redis.pipeline.PipelineOpts.ABExcuteCommands;
import com.huitone.smspfm.socket.client.SmsClient;
import com.huitone.smspfm.socket.protocol.sgip.constant.CommandID;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.IBody;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report2;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;

/**   
* @Title ReportResendManager.java 
* @Package com.huitone.smspfm.interfacz.delrep.report 
* @Description TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年4月11日 下午5:15:31 
* @version V1.0   
*/
public class SmsReportResendTask extends Thread {
	
	private static final ExtLogger logger = ExtLogger.create(SmsReportResendTask.class);
	
	private boolean toSleep = false;
	
	private long sleepTime = 1000;
	
	private int poolSize = (int) Math.ceil(AppStaticConfigs.smspfmRptPoolSize / 2);
	
	private ExecutorService executorService = new ThreadPoolExecutor(poolSize, 
			poolSize, 1L, TimeUnit.MINUTES, new ArrayBlockingQueue<>(poolSize * 2), new RejectedExecutionHandler() {
		
		@Override
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			if (!executor.isShutdown()) {
				try {
					executor.getQueue().put(r);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	});
	
	private boolean keepGoing = true;

	@Override
	public void run() {
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script = new DefaultRedisScript<>(AppStaticConfigs.luaGetResendRpt, List.class);
		
		List<String> keys = new ArrayList<>();
		keys.add(AppStaticConfigs.iSMRptResendList);
		keys.add(AppStaticConfigs.pSMCacheHash);
		
		@SuppressWarnings("rawtypes")
		RedisScript<List> script2 = new DefaultRedisScript<>(AppStaticConfigs.luaRptSendSucc, List.class);
		
		List<String> keys2 = new ArrayList<>();
		keys2.add(AppStaticConfigs.iSMNotifySerialNoSet);
		keys2.add(AppStaticConfigs.iSMRptRespSerialNoZSet);
		
		while(!this.isInterrupted() && isKeepGoing()) {
			
			executorService.execute(new Runnable() {
				
				@Override
				public void run() {
					
					List<List<String>> lists = AppStaticConfigs.scriptOpts3.getHashList(script, keys, AppStaticConfigs.rptReadPerTime + "");
					
					List<SmsReportForResend> entities = new ArrayList<SmsReportForResend>();
					for (int i = 0 ; i < lists.size() ; i ++) {
						List<String> list = lists.get(i);
						SmsReportForResend entity = SmsReportForResend.parse(list);
						entities.add(entity);
					}
					
					if (entities.isEmpty()) {
						toSleep = true;
						return;
					}
					
					toSleep = false;
					
					List<NotifyHash> nfhs = new ArrayList<NotifyHash>();
					
					AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {
						
						@Override
						public void excute(RedisConnection connection) {
					
							for (SmsReportForResend sr:entities) {
								
								if (sr.isWellInformed()) {
									
									ErrorMng errorMng = ErrorMng.get(sr);
									Integer errorCode = errorMng.getInnerCode();
									String errorMsg = errorMng.getErrMsg();
									int reportflag = sr.getReportFlag();
									String srvNode = sr.getSrvNode();
									String cacheKey = sr.getOrgSerial() + ":" + sr.getPhoneNumber();
									
									if ((0 == reportflag && 0 != errorCode) || (1 == reportflag)) { //reportflag = 0 : 该条消息只有最后出错时要返回状态报告 ; reportflag = 1 该条消息无论最后是否成功都要返回状态报告
											  
										SvrNodeDef svrNodeDef = AppStaticConfigs.svrNodeDefMap.get(srvNode);
										
										if (svrNodeDef != null) {
											
											String acceptReport = svrNodeDef.getAcceptReport();
											
											if (StringUtils.equals(acceptReport, "0") || acceptReport == null) { // 是否向业务端推送报告, 0 推送 1 不推送 2 强制不推送(由于多次连接不成功导致)
												
												SmsClient svrNodeClient = SvrNodeClientsManager.getInstance().getSvrNodeClient(srvNode);
												
												if (svrNodeClient != null) {
													
													String upflag = svrNodeDef.getUpflag(); // 新旧客户端标示
													
													String sendport = StringUtils.toString(sr.getSendPort());
													
													IBody report = null;
													
													if (StringUtils.equals("1", upflag) || StringUtils.equals("2", upflag)) {
														
														if (StringUtils.equals("1", upflag)) {
															report = Report2.build(sr.getOrgSerialPart1(), sr.getOrgSerialPart2(), sr.getOrgSerialPart3(), 
																	sr.getNotifyType(), sr.getPhoneNumber(), errorCode, errorMsg, 
																	sendport,
																	sr.getOrgSrvNode(), sr.getOrgSerialNo());
														} else {
															report = Report.build(sr.getOrgSerialPart1(), sr.getOrgSerialPart2(), 
																	sr.getOrgSerialPart3(), sr.getNotifyType(), sr.getPhoneNumber(), errorCode, errorMsg);
														}
														
														Head head = Head.build(report.length(), CommandID.SGIP_REPORT, sr.getLastSerialPart1(), sr.getLastSerialPart2(), sr.getLastSerialPart3());
														
														NotifyHash notifyHash = new NotifyHash();
														notifyHash.setiSMCacheKey(cacheKey);
														notifyHash.setiRptSnPart1(head.getSerialNoPart1());
														notifyHash.setiRptSnPart2(head.getSerialNoPart2());
														notifyHash.setiRptSnPart3(head.getSerialNoPart3());
														notifyHash.setiSendTimes(sr.getSendTimes() + 1);
														notifyHash.setiNotifyType(sr.getNotifyType());
														notifyHash.setiLastSendTime(new Date());
														
														nfhs.add(notifyHash);
														
														Map<byte[], byte[]> hashes = new HashMap<>();
														if (sr.getNotifyType() == 1) {
															hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.ReportSeq, AppStaticConfigs.stringRedisSerializer.serialize(head.getSerialNo()));
														} else {
															hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.SendFeedSeq, AppStaticConfigs.stringRedisSerializer.serialize(head.getSerialNo()));
														}
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
														
														Message message = new Message(head, report); 
														
														logger.rresend(message.toString());
														
														boolean sendReport = svrNodeClient.sendMsg(message);
														
														if (!sendReport) {
															logger.rfailed("重新提交报告信息失败, 内容:" + sr.toString());
														}
														
													} else {
														
														String desc = "重新提交报告信息,客户端标识upflag为空:";
														
														Map<byte[], byte[]> hashes = new HashMap<>();
														if (sr.getNotifyType() == 1) {
															hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
															hashes.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
														} else {
															hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
															hashes.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
															hashes.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
														}
														connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
													
														connection.lPush(AppStaticConfigs.iSMNotifyEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey + ":" + sr.getNotifyType()));
														
														logger.rfailed(desc + sr.toString());
													}
													
												} else {
													String desc = "重新提交报告信息时根据SrvNode编号" + sr.getSrvNode() + "创建客户端失败:";
															
													Map<byte[], byte[]> hashes = new HashMap<>();
													if (sr.getNotifyType() == 1) {
														hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
														hashes.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
														hashes.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
													} else {
														hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
														hashes.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
														hashes.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
													}
													connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
												
													connection.lPush(AppStaticConfigs.iSMNotifyEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey + ":" + sr.getNotifyType()));
													
													logger.rfailed(desc + sr.toString());
												}
											} else {
												
												if (StringUtils.equals(acceptReport, "2")) {
													logger.rfailed("重新提交报告信息,业务端:" + svrNodeDef.getSvrNode() + "当前ACCEPT_REPORT状态为2,业务端连接不可用,直接写状态报告通知完成队列, 报告内容:" + sr.toString());
												}
												
												// 直接写通知完成队列
												AppStaticConfigs.listOpts3.in(AppStaticConfigs.iSMNotifyEndList, cacheKey + ":" + sr.getNotifyType());
											}
										} else {
											String desc = "重新提交报告信息时没有匹配到SrvNode编号" + sr.getSrvNode() + ":";
													
											Map<byte[], byte[]> hashes = new HashMap<>();
											if (sr.getNotifyType() == 1) {
												hashes.put(AppCanstant.SmsEntity.ReportTime, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
												hashes.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
												hashes.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
											} else {
												hashes.put(AppCanstant.SmsEntity.Notify_send_time, AppStaticConfigs.stringRedisSerializer.serialize(DateTimeUtils.dateToStr2(new Date())));
												hashes.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("1"));
												hashes.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize(desc));
											}
											connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + cacheKey), hashes);
										
											connection.lPush(AppStaticConfigs.iSMNotifyEndListB, AppStaticConfigs.stringRedisSerializer.serialize(cacheKey + ":" + sr.getNotifyType()));
											
											logger.rfailed(desc + sr.toString());
										}
		
									} else { // 2 该条消息不需要返回状态报告
										
										// 直接写通知完成队列
										AppStaticConfigs.listOpts3.in(AppStaticConfigs.iSMNotifyEndList, cacheKey + ":" + sr.getNotifyType());
									}
									
								} else {
									logger.rfailed("重新提交报告信息时读取的状态报告信息不全:" + sr.toString());
								}
							}
						}
					});
					
					List<String> serialnoList = new ArrayList<>();
					
					// 保存重发报告需要到的信息
					AppStaticConfigs.pipelineOpts2.save(new ABExcuteCommands() {

						@Override
						public void excute(RedisConnection connection) {
							
							for (NotifyHash nfh:nfhs) {
								
								String hkey = AppStaticConfigs.pSMNotifySerialNo + ":" + nfh.getSerialNo();
								
								byte[] hkeyb = AppStaticConfigs.stringRedisSerializer.serialize(hkey);
								
								connection.hMSet(hkeyb, nfh.getMapB());
								connection.expire(hkeyb, 86400);
								
								serialnoList.add(nfh.getSerialNo() + ":" + (System.currentTimeMillis() + AppStaticConfigs.rptResendInterval));
							}
						}
					});
					
					// 保存状态报告序列号的同时判断是否收到响应消息 
					List<List<String>> nfhstrlist = AppStaticConfigs.scriptOpts2.getHashList(script2, keys2, serialnoList.toArray());
					
					List<NotifyHash> nfhentities = new ArrayList<NotifyHash>();
					for (int i = 0 ; i < nfhstrlist.size() ; i ++) {
						List<String> tmplist = nfhstrlist.get(i);
						NotifyHash entity = NotifyHash.parseStr(tmplist);
						nfhentities.add(entity);
					}
					
					if (!nfhentities.isEmpty()) {
						
						AppStaticConfigs.pipelineOpts3.save(new ABExcuteCommands() {

							@Override
							public void excute(RedisConnection connection) {
								
								List<byte[]> notifyendList = new ArrayList<>();
								
								for (NotifyHash nfh:nfhentities) {
									
									int iNotifyType = nfh.getiNotifyType();
									
									Map<byte[], byte[]> map = new HashMap<>();
									
									if (iNotifyType == 1) {
										map.put(AppCanstant.SmsEntity.ReportResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
										map.put(AppCanstant.SmsEntity.ReportResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
									} else {
										map.put(AppCanstant.SmsEntity.SendFeedbackResult, AppStaticConfigs.stringRedisSerializer.serialize("0"));
										map.put(AppCanstant.SmsEntity.SendFeedbackResultDesc, AppStaticConfigs.stringRedisSerializer.serialize("success"));
									}
									
									connection.hMSet(AppStaticConfigs.stringRedisSerializer.serialize(AppStaticConfigs.pSMCacheHash + ":" + nfh.getiSMCacheKey()), map);
									
									notifyendList.add(AppStaticConfigs.stringRedisSerializer.serialize(nfh.getiSMCacheKey() + ":" + iNotifyType));
								}
								
								if (!notifyendList.isEmpty()) {
									byte[][] bs = notifyendList.toArray(new byte[1][]);
									connection.lPush(AppStaticConfigs.iSMNotifyEndListB, bs);
								}
							}
						});
					}
				}
			});
			
			try {
				if (toSleep) {
					Thread.sleep(sleepTime);
				}
			} catch (InterruptedException e) {}
		}
		
		if (executorService != null && !executorService.isShutdown()) {
			executorService.shutdown();
		}
		
		logger.info("上行报告超时重发任务退出!");
	}
	
	/**
	* @Title: ReportResendManager.java 
	* @Package com.huitone.smspfm.interfacz.delrep.report 
	* @Description: 由于程序Crash或者其它原因,造成在重发过程中可能会出现的SMNotifySerialNoSet集合序列被删除,但是消息却没有发送成功的情况,这时应该根据SMNotifySerialNoPrefix哈希结构的数据修复SMNotifySerialNoSet.
	* @author ybliang   
	* @date 2017年5月31日 下午5:15:12 
	* @version V1.0
	 */
	class ResendSetWatchDog extends Thread {
		
		private long maxInteval = 600000;

		@Override
		public void run() {
			
			while(!this.isInterrupted()) {
				
				Set<String> findKeys = AppStaticConfigs.keyOpts.findKeys(AppStaticConfigs.pSMNotifySerialNo+":*");
				
				for (String key:findKeys) {
					
					NotifyHash notifyHash = NotifyHash.parse(AppStaticConfigs.hashOpts.get(key, NotifyHash.getColumns()));
					
					if (notifyHash.isExists()) {
						
						String serialNo = notifyHash.getSerialNo();
						Date iLastSendTime = notifyHash.getiLastSendTime();
						
						if (iLastSendTime != null && !AppStaticConfigs.zsetOpts.isMember(AppStaticConfigs.iSMNotifySerialNoSet, serialNo)) {
							long lastTime = iLastSendTime.getTime();
							long currTime = System.currentTimeMillis();
							long inteval = currTime - lastTime;
							if (inteval > maxInteval) {
								AppStaticConfigs.zsetOpts.add(key, serialNo, currTime);
							}
						}
					}
				}
				
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					break;
				}
			}
		}
	}

	public boolean isKeepGoing() {
		return keepGoing;
	}

	public void setKeepGoing(boolean keepGoing) {
		this.keepGoing = keepGoing;
	}

	@Override
	public void destroy() {
		this.setKeepGoing(false);
	}
}
